import csv
import os
import time
from subprocess import Popen
from typing import Dict
from uuid import uuid1

import yaml
from kafka import KafkaProducer, KafkaAdminClient, KafkaConsumer
import sys, getopt
import json


def listen_kafka(bootstrap_servers, output_filename, input_topic, output_topic):
    listen_kafka_timeout_min = 40

    consumer = KafkaConsumer(
        input_topic, 
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset='earliest',
        consumer_timeout_ms = listen_kafka_timeout_min * 60 * 1000
    )
    print("listen_kafka started. timeout threshold: " + str(listen_kafka_timeout_min) + " minutes")
    print("input topic :" + input_topic)
    print("output topic :" + output_topic)

    received_message = False
    for message in consumer:
        received_message = True
        break
    
    if received_message:
        print("detected message in input topic. listen_kafka triggered.")
    else:
        print("listen_kafka did not detect message in input topic before timeout. listen_kafka abandoned.")
        return
    time.sleep(10)

    consumer = KafkaConsumer(
        input_topic, 
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset='earliest',
        consumer_timeout_ms = 1000
    )
    input_time = {}
    for message in consumer:
        input_time[int(message.value.decode().split(',')[0])] = message.timestamp
    print('received ' + str(len(input_time)) + ' messages from input topic.')
    time.sleep(10)

    consumer = KafkaConsumer(
        output_topic, 
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset='earliest',
        consumer_timeout_ms = 1000
    )
    output_time = {}
    output_label = {}
    for message in consumer:
        # line = '1,"{""data"":[0.0],""shape"":[]}"'
        line = message.value.decode().strip()
        uid = int(line.split(',')[0])
        output_time[uid] = message.timestamp
        output_label[uid] = int(json.loads(','.join(line.split(',')[1:])[1:-1].replace('""','"'))['data'][0])
    print('received ' + str(len(output_time)) + ' messages from output topic.')
    
    resultf = open(output_filename, 'w+')
    for uid in input_time:
        if uid not in output_label or uid not in output_time:
            continue
        resultf.writelines(['{},{},{},{}\n'.format(uid, input_time[uid], output_time[uid], output_label[uid])])
    print('kafka messages have been written to ' + output_filename)


if __name__ == '__main__':
    opts, args = getopt.getopt(sys.argv[1:],"",["output_filename=","input_topic=","output_topic=","server="])  
    mydict = dict(opts)
    output_filename = mydict.get('--output_filename', '')
    input_topic = mydict.get('--input_topic', '')
    output_topic = mydict.get('--output_topic', '')
    bootstrap_servers = mydict.get('--server', '')
    bootstrap_servers = bootstrap_servers.split(',')
    listen_kafka(bootstrap_servers, output_filename, input_topic, output_topic)
